package com.bw.app.dwd;

import com.bw.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author y3068
 * @Classname Test1
 * @Description TODO
 * @Date 2025/7/3 10:07
 * @Version V1.0
 */

public class test {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        tEnv.executeSql("create table dwd_ub(" +
                "log_id string," +
                "user_id string," +
                "item_id string," +
                "behavior_type string," +
                "behavior_time string," +
                "search_keyword string," +
                "session_id string," +
                "stay_duration string," +
                "page_url string" +
                ")" + MyKafkaUtil.getKafkaDDL("user_behavior_topic", "a"));

//        tEnv.executeSql("select * from dwd_ub").print();

        tEnv.executeSql("create table dwd_ia(" +
                "item_id string," +
                "item_name string," +
                "category_id string," +
                "category_path string," +
                "keywords string," +
                "description string," +
                "price double," +
                "is_maternal string," +
                "create_time string" +
                ")" + MyKafkaUtil.getKafkaDDL("item_attributes_topic", "b"));

//        tEnv.executeSql("select * from dwd_ia").print();
//
        tEnv.executeSql("create table dwd_ua( " +
                "log_id string," +
                "user_id string," +
                "item_id string," +
                "item_name string," +
                "behavior_type string," +
                "behavior_time string," +
                "search_keyword string," +
                "session_id string," +
                "stay_duration string," +
                "page_url string," +
                "category_id string," +
                "category_path string," +
                "keywords string," +
                "description string," +
                "price double," +
                "is_maternal string," +
                "create_time string" +
                ")"+ MyKafkaUtil.getKafkaSinkDDL("dwd_ua"));
//
        Table result = tEnv.sqlQuery("select " +
                "ub.log_id," +
                "ub.user_id," +
                "ub.item_id," +
                "ia.item_name," +
                "ub.behavior_type, " +
                "ub.behavior_time, " +
                "ub.search_keyword, " +
                "ub.session_id, " +
                "ub.stay_duration, " +
                "ub.page_url, " +
                "ia.category_id, " +
                "ia.category_path, " +
                "ia.keywords, " +
                "ia.description, " +
                "ia.price, " +
                "ia.is_maternal, " +
                "ia.create_time " +
                "from dwd_ub ub " +
                "join dwd_ia ia on ub.item_id = ia.item_id");

        tEnv.createTemporaryView("result_table", result);
//        tEnv.sqlQuery("select  * from result_table").execute().print();

        tEnv.executeSql("insert into  dwd_ua select * from result_table");
        tEnv.sqlQuery("select * from dwd_ua").execute().print();

        env.execute();
    }
}
